package org.jordens.jdbcspy.sync;

import org.jordens.jdbcspy.jmx.*;

import javax.management.*;
import javax.management.remote.*;
import java.util.*;
import java.util.concurrent.*;

/**
 * A JMXDataStream provides JDBCSpyRecord events to registered DataUpdaters.
 * 
 * @author Adam Jordens
 */
public class JMXDataStreamer extends AbstractDataStreamer
{
    private final JDBCSpyMBean jdbcSpyMBean;

    private List<JDBCSpyRecord> jdbcSpyRecords =
            new CopyOnWriteArrayList<JDBCSpyRecord>();

    private final Timer refreshTimer = new Timer();

    /**
     * @param connector JMXConnector connection to JDBCSpy MBean
     */
    public JMXDataStreamer(JMXConnector connector)
    {
        if (connector == null)
        {
            throw new NullPointerException("connector cannot be null");
        }

        try
        {
            MBeanServerConnection remote =
                connector.getMBeanServerConnection();

            this.jdbcSpyMBean = (JDBCSpyMBean)
                MBeanServerInvocationHandler.newProxyInstance(
                        remote,
                        new ObjectName(JDBCSpyMBean.NAME),
                        JDBCSpyMBean.class,
                        true);
        }
        catch (Exception e)
        {
            throw new IllegalStateException(e);
        }
    }

    /**
     * {@inheritDoc}
     */
    public void notifyDataUpdaters()
    {
        // Take a local copy of the records that currently need to be published
        // These records will be removed from the publishing stack after sending
        List<JDBCSpyRecord> jdbcSpyRecordsToUpdate =
                    new ArrayList<JDBCSpyRecord>(jdbcSpyRecords);
        try
        {
            for (DataUpdater dataUpdater : getRegisteredDataUpdaters())
            {
                dataUpdater.update(jdbcSpyRecordsToUpdate);
            }
        }
        finally
        {
            jdbcSpyRecords.removeAll(jdbcSpyRecordsToUpdate);
        }
    }

    /**
     * {@inheritDoc}
     */
    public void start()
    {
        refreshTimer.schedule(new RefreshTask(), new Date(), 1000);
    }

    /**
     * Stop the publishing mechanism of this JMXDataStream. 
     */
    public void stop()
    {
        refreshTimer.cancel();
    }

    /**
     * Task that periodically queries the JDBCSpyMBean for statistical information.
     */
    private class RefreshTask extends TimerTask
    {
        private long lastQueryCount = 0;
        private long lastSQLExecutionTime = 0;
        private long lastCheckTimestamp = 0;

        private long numberOfRefreshes = 0;
        
        /**
         * {@inheritDoc}
         */
        public void run()
        {
            long checkTimeStamp = System.currentTimeMillis();

            List<QueryStatistic> updatedQueryStatistics =
                    new ArrayList<QueryStatistic>();

            long queryCount = 0;
            long sqlExecutionTime = 0;

            Collection<QueryStatistic> queryStatistics = jdbcSpyMBean.listQueryStatistics();
            for (QueryStatistic queryStatistic : queryStatistics)
            {
                if (queryStatistic.getLastExecution() > lastCheckTimestamp)
                {
                    updatedQueryStatistics.add(queryStatistic);
                }

                queryCount += queryStatistic.getCount();
                sqlExecutionTime += queryStatistic.getSQLExecutionTime();
            }

            long diffQueryCount = queryCount - lastQueryCount;
            long diffSQLExecutionTime = sqlExecutionTime - lastSQLExecutionTime;

            if (numberOfRefreshes > 0)
            {
                jdbcSpyRecords.add(
                        new JDBCSpyRecord(
                                updatedQueryStatistics,
                                diffSQLExecutionTime,
                                diffQueryCount));

                notifyDataUpdaters();
            }

            lastCheckTimestamp = checkTimeStamp;
            lastQueryCount = queryCount;
            lastSQLExecutionTime = sqlExecutionTime;

            numberOfRefreshes++;
        }
    }

}
